/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lvyh.lightframe.tracer.core;

import com.lvyh.lightframe.tracer.core.mdc.MDCSpanExtension;
import com.lvyh.lightframe.tracer.core.reporter.Reporter;
import com.lvyh.lightframe.tracer.core.util.StringUtils;
import io.opentracing.Span;
import io.opentracing.SpanContext;
import io.opentracing.tag.Tag;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * A span represents the logical operation unit with start time and execution time in the system.
 * Span establishes logical causality through nesting or sequential arrangement,
 * and then constructs the whole call link (tracer) through this relationship
 */
@Data
public class SimpleSpan implements Span, SpanContext {

    private static final Logger logger = LoggerFactory.getLogger(SimpleSpan.class);

    private static final long NO_TIME = 0L;

    public static final int MAX_LAYER = 100;

    private Object dispatcher;

    private Map<String, String> baggageItems;

    private SimpleTracer simpleTracer;

    private List<SpanReferenceRelation> spanReferences;

    private long startTime;

    private long endTime = -1;

    private String operationName;

    private SimpleSpanContext simpleSpanContext;

    private Map<String, String> tags = new LinkedHashMap<>();

    private SimpleSpan parentSpan = null;

    public SimpleSpan(Object dispatcher, Iterable<Map.Entry<String, String>> baggageItems) {
        this.dispatcher = dispatcher;
        this.baggageItems = new HashMap<String, String>();
        if (baggageItems != null) {
            for (Map.Entry<String, String> baggageItem : baggageItems) {
                this.baggageItems.put(baggageItem.getKey(), baggageItem.getValue());
            }
        }
    }

    public SimpleSpan(SimpleTracer simpleTracer, long startTime, String operationName,
                      SimpleSpanContext simpleSpanContext, Map<String, ?> tags) {
        this(simpleTracer, startTime, null, operationName,
                simpleSpanContext != null ? simpleSpanContext : SimpleSpanContext.rootStart(), tags);
    }

    public SimpleSpan(SimpleTracer simpleTracer, long startTime,
                      List<SpanReferenceRelation> spanReferences,
                      String operationName, SimpleSpanContext simpleSpanContext,
                      Map<String, ?> tags) {
        this.simpleTracer = simpleTracer;
        this.startTime = startTime;
        this.spanReferences = spanReferences != null ? new ArrayList<>(spanReferences) : null;
        this.operationName = operationName;
        this.simpleSpanContext = simpleSpanContext;
        this.setTags(tags);
    }

    public SimpleSpan getParentSpan() {
        if (this.parentSpan != null) {
            return this.parentSpan.getThisAsParentWhenExceedLayer();
        }
        return null;
    }

    public void setParentSpan(SimpleSpan parentSpan) {
        this.parentSpan = parentSpan;
    }

    public SimpleSpan getThisAsParentWhenExceedLayer() {
        final SimpleSpan parent;
        String rpcId = this.simpleSpanContext.getSpanId();
        if (StringUtils.countMatches(rpcId, '.') + 1 > MAX_LAYER) {
            SimpleSpanContext parentSpanContext = SimpleSpanContext.rootStart();
            parent = new SimpleSpan(this.simpleTracer, System.currentTimeMillis(), this.operationName, parentSpanContext, null);
        } else {
            parent = this;
        }

        return parent;
    }

    private void setTags(Map<String, ?> tags) {
        if (tags == null || tags.size() <= 0) {
            return;
        }
        for (Map.Entry<String, ?> entry : tags.entrySet()) {
            String key = entry.getKey();
            if (StringUtils.isBlank(key)) {
                continue;
            }
            Object value = entry.getValue();
            if (value == null) {
                continue;
            }
            if (value instanceof String) {
                this.setTag(key, (String) value);
            } else if (value instanceof Boolean) {
                this.setTag(key, (Boolean) value);
            } else if (value instanceof Number) {
                this.setTag(key, (Number) value);
            } else {
                logger.error("Span tags unsupported type [" + value.getClass() + "]");
            }
        }
    }

    public SimpleSpan considerStart(long time) {
        if (time == NO_TIME) {
            return this;
        } else {
            return start(time);
        }
    }

    private SimpleSpan start(long time) {
        return this;
    }

    @Override
    public SpanContext context() {
        return this;
    }

    @Override
    public Iterable<Map.Entry<String, String>> baggageItems() {
        return baggageItems.entrySet();
    }

    @Override
    public void finish() {
        this.finish(System.currentTimeMillis());
    }

    @Override
    public void finish(long finishMicros) {
        this.setEndTime(endTime);
        this.simpleTracer.reportSpan(this);
        MDCSpanExtension.logSpan(this);
    }

    @Override
    public Span setTag(String key, String value) {
        if (StringUtils.isBlank(key) || StringUtils.isBlank(value)) {
            return this;
        }
        this.tags.put(key, value);
        this.simpleSpanContext.getBaggage().put(key, value);
        return this;
    }

    @Override
    public Span setTag(String key, boolean value) {
        return this;
    }

    @Override
    public Span setTag(String key, Number value) {
        return this;
    }

    @Override
    public <T> Span setTag(Tag<T> key, T value) {
        if (key != null && value != null) {
            setTag(key.getKey(), value.toString());
        }
        return this;
    }

    @Override
    public Span log(Map<String, ?> fields) {
        return log(System.currentTimeMillis(), fields);
    }

    @Override
    public Span log(long timestampMicroseconds, Map<String, ?> fields) {
        for (Map.Entry<String, ?> entry : fields.entrySet()) {
            log(timestampMicroseconds, entry.getKey(), entry.getValue());
        }
        return this;
    }

    @Override
    public Span log(String event) {
        return log(System.currentTimeMillis(), event, event);
    }

    @Override
    public Span log(long timestampMicroseconds, String event) {
        return log(timestampMicroseconds, event, event);
    }

    @Override
    public Span setBaggageItem(String key, String value) {
        baggageItems.put(key, value);
        return this;
    }

    @Override
    public String getBaggageItem(String key) {
        return baggageItems.get(key);
    }

    @Override
    public Span setOperationName(String operationName) {
        return this;
    }

    private Span log(long timestampMicroseconds, String eventName, Object payload) {
        if (payload == null) {
            return this;
        }
        return setTag("log." + timestampMicroseconds + "." + eventName, payload.toString());
    }

    @Override
    public String toSpanId() {
        return "";
    }

    @Override
    public String toTraceId() {
        return "";
    }

}
